package com.llx.datastream
import org.apache.flink.api.scala._
import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.runtime.state.filesystem.FsStateBackend
import org.apache.flink.streaming.api.checkpoint.ListCheckpointed
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
import org.apache.flink.streaming.api.{CheckpointingMode, TimeCharacteristic}
import org.apache.flink.streaming.api.functions.source.RichSourceFunction
import org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**setStreamTimeCharacteristic : 设置window数据收集具体用的时间戳 是什么特性
  * AssignerWithPeriodicWatermarks : 设置水位线 时间戳
  * ListCheckpointed : 在apply中执行具体快照内容
  *
  * 例子
  * 需求
  * 假定用户需要每隔1秒钟需要统计4秒中窗口中数据的量，然后对统计的结果值进行checkpoint处理
  * 数据规划
  * 1.使用自定义算子每秒钟产生大约10000条数据。
  *
  * 2.产生的数据为一个四元组(Long，String，String，Integer)—------(id,name,info,count)。
  *
  * 3.数据经统计后，统计结果打印到终端输出。
  *
  * 4.打印输出的结果为Long类型的数据。
  *
  * 开发思路
  *
  * source算子每隔1秒钟发送10000条数据，并注入到Window算子中。
  * window算子每隔1秒钟统计一次最近4秒钟内数据数量。
  * 每隔1秒钟将统计结果打印到终端
  * 每隔6秒钟触发一次checkpoint，然后将checkpoint的结果保存到HDFS中。
  * //发送数据形式
  * case class SEvent(id: Long, name: String, info: String, count: Int)
  * 
  */
object CheckPointExample {

  def main(args: Array[String]): Unit ={
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //设置保存位置
    env.setStateBackend(new FsStateBackend("file:///D:\\checkpoint\\"))
    //设置checkpoint模式
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
    //设置checkpoint间隔
    env.getCheckpointConfig.setCheckpointInterval(6000)
    // Flink DataStream程序的第一部分通常设置基本的时间特性(base time characteristic)。
    // 该设置定义数据流源的行为方式(例如，它们是否产生时间戳)，
    // 以及窗口操作如KeyedStream.timeWindow(Time.seconds(30))应使用哪一类型时间，是事件时间还是处理时间等。
    //设置streamtime特性 （Characteristic） KeyedStream.timeWindow(Time.seconds(30))应使用哪一类型时间
    // EventTime事件时间  :应用与事件的创建时间
    // IngestionTime提取时间  :应用与事件在flink抽取成dataSource 的时间
    // ProcessingTime处理时间 ：应用与事件在flink中Window窗口的处理当前元素的时间
    /**eg：
      * 统计业务时间上的每小时的点击量，所以要基于 EventTime 来处理。
      * 现在按照 EventTime 模式进行处理，Flink 默认使用 ProcessingTime处理，所以我们要显式设置下。
      * env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
      *
      */
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    // 应用逻辑  添加自定义源
    val source: DataStream[SEvent] = env.addSource(new SEventSourceWithChk)
    //为 自定义源一般都要添加时间戳，后续有Window操作 添加时间戳 设置水位线 为window
    /**
      * 由于我们的数据源的数据已经经过整理，没有乱序，即事件的时间戳是单调递增的，
      * 所以可以将每条数据的业务时间就当做 Watermark。
      * 这里我们用AscendingTimestampExtractor 来实现时间戳的抽取和 Watermark 的生成。
      *
      *注意：
      * 真 实业务场景一般都是存在乱序的， 所以一般使用BoundedOutOfOrdernessTimestampExtractor
      *
      *
      */
    source.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks[SEvent] {
      // 设置watermark
      override def getCurrentWatermark: Watermark = {
        new Watermark(System.currentTimeMillis())
      }
      // 给每个元组打上时间戳
      override def extractTimestamp(t: SEvent, l: Long): Long = {
        System.currentTimeMillis()
      }
    })
      .keyBy(0)
      //窗口大小4  每秒滑动一次
      .window(SlidingEventTimeWindows.of(Time.seconds(4), Time.seconds(1)))
      //收集的窗口执行 WindowStatisticWithChk:执行 每秒执行WindowStatisticWithChk.apply   每6秒执行：ListCheckpointed快照
      .apply(new WindowStatisticWithChk)
      .print()
    env.execute()
  }

  //该段代码是window算子的代码，每当触发计算时统计窗口中元组数量。
  class WindowStatisticWithChk extends WindowFunction[SEvent, Long, Tuple, TimeWindow] with ListCheckpointed[UDFState]{
    private var total = 0L

    // window算子的实现逻辑，即:统计window中元组的数量
    override def apply(key: Tuple, window: TimeWindow, input: Iterable[SEvent], out: Collector[Long]): Unit = {
      var count = 0L
      for (event <- input) {
        count += 1L
      }
      total += count
      out.collect(count)
    }
    // 从自定义快照中恢复状态 restore回复
    override def restoreState(state: java.util.List[UDFState]): Unit = {
      val udfState = state.get(0)
      total = udfState.getState
    }

    // 制作自定义状态快照 获取要snapshot的数据 list
    override def snapshotState(checkpointId: Long, timestamp: Long): java.util.List[UDFState] = {
      val udfList: java.util.ArrayList[UDFState] = new java.util.ArrayList[UDFState]
      val udfState = new UDFState
      udfState.setState(total)
      udfList.add(udfState)
      udfList
    }
  }

  /**该数据在算子制作快照时用于保存到目前为止算子记录的数据条数。
    * 用户自定义状态
    */
  class UDFState extends Serializable{
    private var count = 0L
    // 设置用户自定义状态
    def setState(s: Long) = count = s
    // 获取用户自定状态
    def getState = count
  }

  

  /**
    *
    * @param id
    * @param name
    * @param info
    * @param count
    */
  case class SEvent(id: Long, name: String, info: String, count: Int)


  /**
    *
    */
  class SEventSourceWithChk extends RichSourceFunction[SEvent]{
    private var count = 0L
    private var isRunning = true
    private val alphabet = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWZYX0987654321"
    // 任务取消时调用
    override def cancel(): Unit = {
      isRunning = false
    }
    // source算子的逻辑，即:每秒钟向流图中注入10000个元组
    override def run(sourceContext: SourceContext[SEvent]): Unit = {
      while(isRunning) {
        for (i <- 0 until 10000) {
          sourceContext.collect(SEvent(1, "hello-"+count, alphabet,1))
          count += 1L
        }
        Thread.sleep(1000)
      }
    }
  }
 
}
